package com.lvmama.task.service.impl;

import com.lvmama.comm.utils.MemcachedUtil;
import com.lvmama.task.common.service.TaskService;
import com.lvmama.task.common.service.TaskScheduleService;
import com.lvmama.task.common.vo.Task;
import com.lvmama.task.common.vo.TaskTick;
import com.lvmama.task.service.dao.TaskTickDAO;


import java.util.*;

/**
 * Created by troy on 2016/7/4.
 * kouhongyu@163.com
 */
public class TaskScheduleServiceImpl implements TaskScheduleService {

    private TaskTickDAO taskTickDAO;
    private TaskService taskService;
    /**
     * 正在选举的缓存KEY
     */
    private static final String TASK_LEADER_ELECTION_RUNNING = "task.leader.election.running";
    /**
     * 缓存时长
     */
    private static final int TASK_LEADER_ELECTION_RUNNING_WHEN_LONG = 8;
    /**
     * 持续多长时间没有勾点将被清理(秒)
     */
    private static final Integer TICK_POINT_OLD_DURATION = -10;

    @Override
    public void tick(String jobNode, String owner, String ip, String info) {
        Map<String, Object> nodeOwnerMap = new HashMap<String, Object>();
        nodeOwnerMap.put("jobNode", jobNode);
        nodeOwnerMap.put("owner", owner);
        TaskTick taskTick = taskTickDAO.getTaskTickByNodeOwner(nodeOwnerMap);

        if (taskTick == null) {
            taskTick = new TaskTick();

            taskTick.setJobNode(jobNode);
            taskTick.setOwner(owner);
            taskTick.setTickPoint(new Date());
            taskTick.setActive(TaskTick.AFFIRM.Y);
            taskTick.setLeader(TaskTick.AFFIRM.N);
            taskTick.setIp(ip);
            taskTick.setInfo(info);

            taskTickDAO.insertTaskTick(taskTick);

        } else {
            taskTick.setTickPoint(new Date());
            taskTick.setActive(TaskTick.AFFIRM.Y);
            taskTick.setIp(ip);
            taskTick.setInfo(info);

            taskTickDAO.updateTaskTick(taskTick);
        }
    }

    @Override
    public void election(String jobNode, String owner) {
        boolean electionRunning = MemcachedUtil.getInstance().get(TASK_LEADER_ELECTION_RUNNING) != null;

        if (!electionRunning) {
            Map<String, Object> nodeOwnerMap = new HashMap<String, Object>();
            nodeOwnerMap.put("jobNode", jobNode);
            nodeOwnerMap.put("owner", owner);
            TaskTick eleTickNode = taskTickDAO.getTaskTickByNodeOwner(nodeOwnerMap);

            if (eleTickNode != null) {
                /**
                 * 清理不活跃的节点
                 */
                clearInactive(owner);
                /**
                 * 清理旧的死进程节点记录(一天以前的)
                 */
                clearDieThreadNode(owner);

                Map<String, Object> paramMap = new HashMap<String, Object>();
                paramMap.put("active", TaskTick.AFFIRM.Y);
                paramMap.put("owner", owner);

                List<TaskTick> taskTickList = taskTickDAO.getTaskTickList(paramMap);
                if (taskTickList != null) {
                    for (int i = 0; i < taskTickList.size(); i++) {
                        TaskTick taskTick = taskTickList.get(i);

                        if (i == 0) {
                            /**
                             * 第一个活跃的节点选为调度器
                             */
                            taskTick.setLeader(TaskTick.AFFIRM.Y);
                        } else {
                            taskTick.setLeader(TaskTick.AFFIRM.N);
                        }
                        taskTickDAO.updateTaskTick(taskTick);
                    }
                }

                MemcachedUtil.getInstance().set(TASK_LEADER_ELECTION_RUNNING, TASK_LEADER_ELECTION_RUNNING_WHEN_LONG, jobNode);
            }
        }
    }

    @Override
    public void schedule(String jobNode, String owner) {
        Map<String, Object> nodeOwnerMap = new HashMap<String, Object>();
        nodeOwnerMap.put("jobNode", jobNode);
        nodeOwnerMap.put("owner", owner);
        TaskTick taskTick = taskTickDAO.getTaskTickByNodeOwner(nodeOwnerMap);
        /**
         * 如果当前节点是调度器,就进行任务调度分配
         */
        if (taskTick != null && TaskTick.AFFIRM.Y.equals(taskTick.getLeader())) {

            /**
             * 活跃的节点列表
             */
            Map<String, Object> tickParamMap = new HashMap<String, Object>();
            tickParamMap.put("active", TaskTick.AFFIRM.Y);
            tickParamMap.put("owner", owner);
            List<TaskTick> activeNodeList = taskTickDAO.getTaskTickList(tickParamMap);

            /**
             * 启用的任务
             */
            Map<String, Object> taskParamMap = new HashMap<String, Object>();
            taskParamMap.put("owner", owner);
            taskParamMap.put("available", Task.TASK_AVAILABLE.ENABLE);
            taskParamMap.put("_startRow", 1);
            taskParamMap.put("_endRow", 10000);

            List<Task> taskList = taskService.getTaskList(taskParamMap);

            /**
             * 调度
             */
            int i = 0;
            for (Task task : taskList) {
                /**
                 * 轮询分配节点
                 */
                String newNode = activeNodeList.get(Math.abs(i++) % activeNodeList.size()).getJobNode();
                if (!newNode.equals(task.getJobNode())) {
                    task.setJobNode(newNode);
                    taskService.saveTaskRuntime(task);
                }
            }
        }
    }

    /**
     * 清理不活跃的节点
     */
    private void clearInactive(String owner) {
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.SECOND, TICK_POINT_OLD_DURATION);

        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("tickPointOldDuration", calendar.getTime());
        paramMap.put("owner", owner);

        /**
         * 10秒不活跃将被清理
         */
        taskTickDAO.clearInactiveNode(paramMap);
    }

    /**
     * 清理旧的死进程节点记录(一天以前的)
     */
    private void clearDieThreadNode(String owner) {
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.DAY_OF_MONTH, -1);

        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("tickPointOldDuration", calendar.getTime());
        paramMap.put("owner", owner);

        /**
         * 一天以前的不活跃将被删除记录
         */
        taskTickDAO.clearDieThreadNode(paramMap);
    }

    public void setTaskTickDAO(TaskTickDAO taskTickDAO) {
        this.taskTickDAO = taskTickDAO;
    }

    public void setTaskService(TaskService taskService) {
        this.taskService = taskService;
    }
}
